package com.o2o.cleaning.month.platform.ebusiness_plat.meituan

import org.apache.spark.sql._


/**
  * 提取美团新增分类和地址
  *
  * @ Auther: o2o-rd-0008
  * @ Date:   2018/11/2 17:13
  * @ Param:  $param$
  * @ Description:
  */
object SourceShopAddress {


  /**
    * meituan提取新增分类和商品详情中的地址
    *
    * @param spark
    */
  def extract_Address_newAddCategory(spark: SparkSession, detail_sourcePath: String): Unit = {

    /** *******************重要重要重要重要重要重要 ************************************************************/


    //meituan 原有分类路径
    val category_path = Meituan_config.category_path
    //保存本次新增分类保存路径
    val category_newAdd_path = Meituan_config.category_newAdd_path

    //拉取数据中的地址保存路径
    val addr_path = Meituan_config.addr_path_from_detail


    //    =========== 简单的计算 =============

    val frame_0: DataFrame = spark.read.orc(detail_sourcePath)

    val frame = frame_0
      .selectExpr("shopId", "IFNULL(regional_id,'0') regional_id", "IFNULL(latitude,'-1') latitude", "IFNULL(longitude,'-1') longitude"
        , "IFNULL(address,'0') address", "IFNULL(province,'0') province", "IFNULL(city,'0') city", "IFNULL(county,'0') county"
        , "mtWmPoiId", "shopName", "title", "good_id", "shopId", "priceText", "sellCount", "rootCategoryId", "rootCategoryName", "categoryId",
        "categoryName")

    // 保存地址数据
    frame.select("shopId", "regional_id", "latitude", "longitude", "address", "province", "city", "county")
    //临时注释
     .dropDuplicates().coalesce(1).write.json(addr_path)


    val data_cate: Dataset[Row] = frame.select("rootCategoryId", "rootCategoryName", "categoryId", "categoryName", "title")
      .dropDuplicates().coalesce(1)

    //新增分类
    // 这里的category_path原有的分类路径指的是上个月的路径吗？   新增分类指的是相比较上个月的分类，这个月增加了哪些分类吗？
    //6月美团计算出的新增分类结果很多，是因为使用categoryId来判断新增分类，相同的分类可能5月份和6月份的categoryId不同。
    val isHaveNewCate_row: Dataset[Row] = data_cate.select("categoryId").except(spark.read.json(category_path).select("categoryId"))
      .coalesce(1)
      .where("categoryId !=-1")
    //临时注释
     isHaveNewCate_row.join(data_cate, Seq("categoryId"), "left").coalesce(1)
        .write.option("header", true).csv(category_newAdd_path)
    val quantity: Long = isHaveNewCate_row.count()
    if (quantity > 0) {
      println("美团存在新增分类" + quantity)
      //合并保存 ，不进行非空判断，确保每月都有新增分类文件生成 （即使文件为空），
      throw new RuntimeException("美团存在新增分类，条数是：" + quantity)
    }


  }

}

